/**
 *	@file	work3.c
 *
 *	@brief	work thread which accept client, 
 *		and recv client request data, parse request data, 
 *		connect to server, send request to server and recv 
 *		reply from server, parse reply, and send reply to 
 *		client. 
 *	
 *	@author	Forrest.zhang	
 *
 *	@date	2008-07-31
 */

#define	_GNU_SOURCE

#include <unistd.h>
#include <signal.h>
#include <assert.h>
#include <stdlib.h>
#include <errno.h>
#include <string.h>
#include <stdio.h>
#include <sys/epoll.h>
#include <fcntl.h>

#include "debug.h"
#include "thread.h"
#include "proxy.h"
#include "session.h"
#include "ip_addr.h"
#include "sock_util.h"
#include "nb_splice.h"
#include "cpu_util.h"

//#define NDEBUG          1

/**
 *	Init work thread, alloc resources.
 *
 *	Return 0 if success, -1 on error.
 */
static int 
_work_init(thread_t *info)
{
	work_t *wi;
	int maxfds = 0;
	int *pair;
	struct epoll_event e;

	assert(info);

	/* alloc work_info */
	wi = malloc(sizeof(work_t));
	if (!wi) {
		ERR("malloc error %s\n", ERRSTR);
		return -1;
	}
	memset(wi, 0, sizeof(work_t));

	maxfds = (g_proxy.max / g_proxy.nwork) + 1;

	wi->max = maxfds;
	
	/* alloc memory for epoll event */
	wi->events = malloc(maxfds * sizeof(struct epoll_event) * 2);
	if (!wi->events) {
		ERR("malloc error %s\n", ERRSTR);
		return -1;
	}
	wi->nevent = maxfds * 2;
	DBG("work[%d] alloc %d epoll events(%p)\n", 
	    info->index, (maxfds * 2), wi->events);

	/* alloc session table */
	wi->sesstbl = session_table_alloc(g_proxy.max, 0);
	if (!wi->sesstbl) {
		ERR("session_table_alloc failed\n");
		return -1;
	}
	DBG("work[%d] alloc session table(%p)\n", info->index, wi->sesstbl);
	
	/* alloc object pool */
	wi->pktpool = objpool_alloc(4096, 1024, 0);
	if (!wi->pktpool) {
		ERR("objpool_alloc failed\n");
		return -1;
	}
	DBG("work[%d] alloc packet pool(%p)\n", info->index, wi->pktpool);

	/* alloc session pool */
	wi->ssnpool = objpool_alloc(sizeof(session_t), 1024, 0);
	if (!wi->ssnpool) {
		ERR("objpool_alloc failed\n");
		return -1;
	}
	DBG("work[%d] alloc session pool(%p)\n", info->index, wi->ssnpool);

	/* create epoll fd */
	wi->epfd = epoll_create(maxfds);
	if (wi->epfd < 0) {
		ERR("epoll_create error %s\n", ERRSTR);
		return -1;
	}
	DBG("work[%d] create epoll fd %d\n", info->index, wi->epfd);

	/* create listen socket */
	wi->httpfd = sk_tcp_server(&g_proxy.httpaddr, 1);
	if (wi->httpfd < 0) {
		ERR("create http listen fd failed\n");
		return -1;
	}
	sk_set_nonblock(wi->httpfd, 1);

	/* add listen fd to recv epoll */
	pair = (int *)&e.data.u64;
	pair[0] = wi->httpfd;
	pair[1] = 0;
	e.events = EPOLLIN;		
	if (epoll_ctl(wi->epfd, EPOLL_CTL_ADD, wi->httpfd, &e)) {
		ERR("epoll add %d error %s\n", wi->httpfd, ERRSTR);
		return -1;
	}
	DBG("work[%d] create http listen fd %d\n", info->index, wi->httpfd);

	info->priv = wi;

	return 0;
}

/**
 *	Release work thread resource which alloced by _work_init()
 *
 *	No return.
 */
static void 
_work_free(thread_t *info)
{
	work_t *wi;

	assert(info);

	wi = info->priv;	
	if (!wi)
		return;

	if (wi->inq) {
		free(wi->inq);
		wi->inq = NULL;
		DBG("work[%d] freed input queue\n", info->index);
	}

	if (wi->inq2) {
		free(wi->inq2);
		wi->inq2 = NULL;
		DBG("work[%d] freed second input queue\n", info->index);
	}

	if (wi->events) {
		free(wi->events);
		wi->events = NULL;
		DBG("work[%d] free epoll event\n", info->index);
	}

	if (wi->sesstbl) {
		session_table_free(wi->sesstbl);
		wi->sesstbl = NULL;
		DBG("work[%d] free session table\n", info->index);
	}

	if (wi->epfd >= 0) {
		close(wi->epfd);
		DBG("work[%d] close send epoll %d\n", 
		    info->index, wi->epfd);
	}

	info->priv = NULL;
	free(wi);
}

/**
 *	Get a packet from pktpool if need, and put it to @q->pkt.
 *
 *	Return pointer to packet_t if success, NULL on error.
 */
static packet_t *
_work_get_packet(thread_t *info, session_t *s, session_queue_t *q)
{
	packet_t *pkt;
	work_t *wi;

	assert(info);
	assert(info->priv);
	assert(s);
	assert(q);

	wi = info->priv;

	pkt = q->pkt;
	if (pkt)
		return pkt;

	pkt = objpool_get(wi->pktpool);
	if (!pkt) {
		ERR("can't get packet from pkt pool\n");
		return NULL;
	}
	s->nalloced++;
	q->pkt = pkt;
	pkt->len = 0;
	pkt->sendpos = 0;
	pkt->pos = 0;
	pkt->capacity = wi->pktpool->objsize - sizeof(packet_t);

	if (q == &s->cliq) {
		FLOW(2, "work[%d]: ssn<%d> client %d get a packet(%p)\n", 
		     info->index, s->id, s->clifd, pkt);
	}
	else {
		FLOW(2, "work[%d]: ssn<%d> server %d get a packet(%p)\n",
		     info->index, s->id, s->svrfd, pkt);
	}

	return pkt;
}

/**
 *	Select a real server for session @s, use round-robin algo.
 *
 *	Return 0 if success, -1 on error.
 */
static int 
_work_get_server(thread_t *info, session_t *s)
{
	static int i = 0;
	work_t *wi;
	struct epoll_event e;
	int fd;
	int wait;
	int *pair;
	int index;

	assert(info);
	assert(info->priv);
	assert(s);

	wi = info->priv;

	if (s->svrfd > 0)
		return 0;

	proxy_stat_lock();
	index = i;
	i = (i + 1) % g_proxy.nrsaddr;
	proxy_stat_unlock();

	fd = sk_tcp_client_nb(&g_proxy.rsaddrs[index], &wait);
	if (fd < 0) {
		s->is_svrerror = 1;
		ERR("connect server failed\n");
		return -1;
	}

	sk_set_nodelay(fd, 1);
	sk_set_quickack(fd, 1);

	/* add it to epoll to wait connect success */
	if (wait) {

		/* add to epoll for write */
		pair = (int *)&e.data.u64;	
		pair[0] = fd;
		pair[1] = s->id;
		e.events = EPOLLOUT;			
		if (epoll_ctl(wi->epfd, EPOLL_CTL_ADD, fd, &e)) {
			s->is_svrerror = 1;
			close(fd);
			ERR("epoll add %d error %s\n", fd, ERRSTR);
			return -1;
		}
		s->is_svrep = 1;
		FLOW(3, "work[%d]: ssn<%d> server %d epoll add write\n",
		     info->index, s->id, fd);

		/* remove client from epoll */
		if (epoll_ctl(wi->epfd, EPOLL_CTL_DEL, s->clifd, NULL)) {
			s->is_clierror = 1;
			ERR("epoll del %d error %s\n", fd, ERRSTR);
			return -1;
		}
		s->is_cliep = 0;
		FLOW(3, "work[%d]: ssn<%d> client %d epoll del read\n",
		     info->index, s->id, s->clifd);

		s->is_svrwait = 1;
		wi->nblocked ++;
		FLOW(1, "work[%d]: ssn<%d> client %d connect to server %d wait\n", 
		     info->index, s->id, s->clifd, fd);
	}
	/* add to epoll if connect success */
	else {
		pair = (int *)&e.data.u64;	
		pair[0] = fd;
		pair[1] = s->id;
		e.events = EPOLLIN;			
		if (epoll_ctl(wi->epfd, EPOLL_CTL_ADD, fd, &e)) {
			ERR("epoll add %d error %s\n", fd, ERRSTR);
			s->is_svrerror = 1;
			close(fd);
			return -1;
		}
		s->is_svrep = 1;
		FLOW(3, "work[%d]: ssn<%d> server %d epoll add read\n",
		     info->index, s->id, fd);

		FLOW(1, "work[%d]: ssn<%d> client %d connect to server %d success\n", 
		     info->index, s->id, s->clifd, fd);
	}

	s->svrfd = fd;

	return 0;
}

static session_t * 
_work_alloc_session(thread_t *info)
{
	work_t *wi;
	session_t *s;

	assert(info);
	assert(info->priv);

	wi = info->priv;

	s = objpool_get(wi->ssnpool);
	if (!s) {
		ERR("objpool_get failed\n");
		return NULL;
	}
	memset(s, 0, sizeof(session_t));
	s->clifd = -1;
	s->svrfd = -1;
	s->id = 0;

	if (session_table_add(wi->sesstbl, s)) {
		ERR("session_table_add failed\n");
		objpool_put(s);
		return NULL;
	}

	return s;
}

/**
 *	Clean session queue @q
 *
 *	No return.
 */
static void 
_work_clean_session(thread_t *info, session_t *s, session_queue_t *q)
{
	packet_t *pkt;
	
	assert(info);
	assert(s);
	assert(q);
	
	pkt = q->pkt;
	if (pkt) {
		s->nalloced--;
		FLOW(2, "work[%d]: ssn<%d> free current packet %p\n", 
		     info->index, s->id, pkt);
		objpool_put(pkt);
	}

	pkt = q->blkpkt;
	if (pkt) {
		s->nalloced--;
		FLOW(2, "work[%d]: ssn<%d> free block packet %p\n", 
		     info->index, s->id, pkt);
		objpool_put(pkt);
	}

	pkt = pktq_out(&q->inq);
	while(pkt) {
		s->nalloced--;
		FLOW(2, "work[%d]: ssn<%d> free inq packet %p\n", 
		     info->index, s->id, pkt);
		objpool_put(pkt);
		pkt = pktq_out(&q->inq);
	}

	pkt = pktq_out(&q->outq);
	while (pkt) {
		s->nalloced--;
		FLOW(2, "work[%d]: ssn<%d> free outq packet %p\n", 
		     info->index, s->id, pkt);
		objpool_put(pkt);
		pkt = pktq_out(&q->outq);
	}
}

/**
 *	Free session @s, release all resource used by @s.
 *
 *	No return.
 */
static void 
_work_free_session(thread_t *info, session_t *s)
{
	work_t *wi;
	int fd;
	int epfd;
	int is_cliclose;
	int is_clierror;
	int is_svrclose;
	int is_svrerror;

	assert(info);
	assert(info->priv);	
	assert(s);

	wi = info->priv;

	if (g_proxy.use_splice) {
		if (s->read_pipe > 0)
			close(s->read_pipe);
		if (s->write_pipe > 0)
			close(s->write_pipe);
	}

	/* close client socket */
	if (s->clifd > 0) {
		fd = s->clifd;
		epfd = wi->epfd;

		if (s->is_cliep && epoll_ctl(epfd, EPOLL_CTL_DEL, fd, NULL)) {
			ERR("epoll del %d error %s\n", fd, ERRSTR);
		}
		
		FLOW(1, "work[%d]: ssn<%d> close client %d\n", 
		     info->index, s->id, fd);
		close(fd);
	}

	/* close server socket */
	if (s->svrfd > 0) {
		fd = s->svrfd;
		epfd = wi->epfd;

		if (s->is_svrep & epoll_ctl(epfd, EPOLL_CTL_DEL, fd, NULL))
			ERR("epoll del %d error %s\n", fd, ERRSTR);

		FLOW(1, "work[%d]: ssn<%d> close server %d\n", 
		     info->index, s->id, fd);

		close(fd);
	}

	_work_clean_session(info, s, &s->cliq);

	_work_clean_session(info, s, &s->svrq);

	assert(s->nalloced == 0);

	/* delete session from session pool */
	FLOW(1, "work[%d]: ssn<%d> deleted\n", info->index, s->id);

	session_table_del(wi->sesstbl, s->id);

	is_cliclose = s->is_cliclose;
	is_svrclose = s->is_svrclose;
	is_clierror = s->is_clierror;
	is_svrerror = s->is_svrerror;

	objpool_put(s);

	proxy_stat_lock();
	g_proxy.stat.nhttplive--;
	g_proxy.stat.nlive--;
	if (is_cliclose)
		g_proxy.stat.ncliclose++;
	if (is_svrclose)
		g_proxy.stat.nsvrclose++;
	if (is_clierror)
		g_proxy.stat.nclierror++;
	if (is_svrerror)
		g_proxy.stat.nsvrerror++;
	proxy_stat_unlock();
}

/**
 *	Accept new client and add to session table.
 *
 */
static int 
_work_accept_client(thread_t *info, int listenfd)
{
	int clifd;
	ip_port_t cliaddr;
	work_t *wi;
	session_t *s;
	int *pair;
	int pipefd[2];
	u_int64_t nlive;
	struct epoll_event e;
	char ipstr1[IP_STR_LEN];
	char ipstr2[IP_STR_LEN];

	assert(info);
	assert(info->priv);
	assert(listenfd > 0);

	wi = (work_t *)info->priv;

	/* accept client */
	clifd = sk_tcp_accept(listenfd, &cliaddr);
	if (clifd < 0) {
		if (errno != EAGAIN) {
			ERR("sk_tcp_accept %d failed: %s\n", 
					listenfd, ERRSTR);
			return -1;
		}
		else {
			return 0;
		}
	}

	proxy_stat_lock();
	g_proxy.stat.naccept++;
	proxy_stat_unlock();

	sk_set_nonblock(clifd, 1);
	sk_set_nodelay(clifd, 1);
	sk_set_quickack(clifd, 1);

	/* check max connection */
	proxy_stat_lock();
	nlive = g_proxy.stat.nlive;
	proxy_stat_unlock();
	if (nlive >= g_proxy.max) {
		close(clifd);
		ERR("work[%d] too many client %lu max %d\n", 
				info->index, nlive, g_proxy.max);
		return -1;
	}

	/* alloc new session */
	s = _work_alloc_session(info);
	if (!s) {
		close(clifd);
		return -1;
	}

	/* add to recv epoll */
	pair = (int *)&e.data.u64;
	pair[0] = clifd;
	pair[1] = s->id;
	e.events = EPOLLIN;
	if (epoll_ctl(wi->epfd, EPOLL_CTL_ADD, clifd, &e)) {
		_work_free_session(info, s);
		close(clifd);
		ERR("epoll add %d error %s\n", clifd, ERRSTR);
		return -1;
	}
	s->is_cliep = 1;
	FLOW(3, "work[%d]: ssn<%d> client %d epoll add read\n",
	     info->index, s->id, clifd);
	s->clifd = clifd;

	/* connect to real server if use_splice */
	if (g_proxy.use_splice || g_proxy.use_nb_splice) {
		if (_work_get_server(info, s)) {
			_work_free_session(info, s);
			return -1;
		}
	}

	/* create pipe for splice */
	if (g_proxy.use_splice) {
		if (pipe2(pipefd, O_NONBLOCK)) {
			ERR("pipe2 error %s\n", ERRSTR);
			_work_free_session(info, s);
			return -1;
		}

		s->read_pipe = pipefd[0];
		s->write_pipe = pipefd[1];

		FLOW(1, "work[%d]: ssn<%d> create splice pipe %d:%d\n",
				info->index, s->id, s->read_pipe, s->write_pipe);
	}

	FLOW(1, "work[%d]: ssn<%d> client %d accepted(%s->%s)\n",
	     info->index, s->id, s->clifd, 
	     ip_port_to_str(&cliaddr, ipstr1, IP_STR_LEN),
	     ip_port_to_str(&g_proxy.httpaddr, ipstr2, IP_STR_LEN));

	/* update proxy statistic */
	proxy_stat_lock();
	g_proxy.stat.nhttp++;
	g_proxy.stat.nhttplive++;
	g_proxy.stat.nlive++;
	proxy_stat_unlock();

	return 0;
}


static int 
_work_check_server(thread_t *info, session_t *s)
{
	work_t *wi;
	struct epoll_event event;
	int ret = 0;
	int fd;
	int *pair;

	assert(info);
	assert(info->priv);
	assert(s);

	wi = info->priv;
	fd = s->svrfd;

	ret = sk_is_connected(fd);
	if (ret <= 0) {
		FLOW(1, "work[%d]: ssn<%d> server %d connect failed\n", 
		     info->index, s->id, fd);
		return -1;
	}
	s->is_svrwait = 0;
	wi->nblocked --;

	/* mod it in epoll(read) */
	pair = (int *)&event.data.u64;
	pair[0] = fd;
	pair[1] = s->id;
	event.events = EPOLLIN;
	if (epoll_ctl(wi->epfd, EPOLL_CTL_MOD, fd, &event)) {
		ERR("epoll mod %d error %s\n", fd, ERRSTR);
		return -1;
	}
	FLOW(3, "work[%d]: ssn<%d> server %d epoll mod write->read\n",
	     info->index, s->id, fd);

	/* add client to epoll(read) */
	pair = (int *)&event.data.u64;
	pair[0] = s->clifd;
	pair[1] = s->id;
	event.events = EPOLLIN;
	if (epoll_ctl(wi->epfd, EPOLL_CTL_ADD, s->clifd, &event)) {
		ERR("epoll add %d error %s\n", s->clifd, ERRSTR);
		return -1;
	}
	FLOW(3, "work[%d]: ssn<%d> client %d epoll add read\n",
	     info->index, s->id, s->clifd);

	FLOW(1, "work[%d]: ssn<%d> client %d connect to server %d success\n", 
	     info->index, s->id, s->clifd, fd);

	return 0;
}


static int 
_work_recvfrom_client(thread_t *info, session_t *s)
{
	packet_t *pkt;
	int fd;
	int n;
	int blen = 0;
	int close = 0;

	assert(info);
	assert(info->priv);
	assert(s);
	assert(s->clifd >= 0);
	
	fd = s->clifd;

	/* get packet to restore recved data */
	pkt = _work_get_packet(info, s, &s->cliq);
	if (!pkt)
		return -1;
	
	blen = pkt->capacity - pkt->len;
	assert(blen > 0);
	n = sk_recv(fd, pkt->data, blen, &close);
	if (n < 0) {
		s->is_clierror = 1;
		ERR("work[%d]: ssn<%d> client %d recv error\n", 
		    info->index, s->id, fd);
		return -1;
	}
	
	pkt->len += n;

	proxy_stat_lock();
	g_proxy.stat.nclirecv += n;
	proxy_stat_unlock();

	FLOW(1, "work[%d]: ssn<%d> client %d recv %d bytes\n", 
	     info->index, s->id, fd, n);

	if (pkt->len == 0) {
		objpool_put(pkt);
		s->cliq.pkt = NULL;
		s->nalloced--;

		FLOW(2, "work[%d]: ssn<%d> client %d free packet(%p)\n", 
		     info->index, s->id, fd, pkt);
	}

	if (close) {
		s->is_cliclose = 1;
		FLOW(1, "work[%d]: ssn<%d> client %d recv closed\n", 
		     info->index, s->id, fd);
	}

	return 0;
}

static int 
_work_recvfrom_server(thread_t *info, session_t *s)
{
	packet_t *pkt;
	int n;
	int blen = 0;
	char *ptr;
	int fd;
	int close = 0;

	assert(info);
	assert(info->priv);
	assert(s);
	assert(s->svrfd >= 0);
	
	fd = s->svrfd;

	pkt = _work_get_packet(info, s, &s->svrq);
	if (unlikely(!pkt)) {
		return -1;
	}

	/* recv data */
	ptr = pkt->data + pkt->recvpos;
	blen = pkt->capacity - pkt->recvpos;
	assert(blen > 0);
	n = sk_recv(fd, ptr, blen, &close);
	if (n < 0) {
		s->is_svrerror = 1;
		ERR("work[%d]: ssn<%d> server %d recv error\n", 
		     info->index, s->id, fd);
		return -1;
	}
	
	pkt->len += n;

	proxy_stat_lock();
	g_proxy.stat.nsvrrecv+=n;
	proxy_stat_unlock();

	if (pkt->len == 0) {
		objpool_put(pkt);
		s->svrq.pkt = NULL;
		s->nalloced--;
		FLOW(2, "work[%d] ssn<%d> server %d free packet(%p)\n", 
		     info->index, s->id, fd, pkt);
	}

	if (close) {
		s->is_svrclose = 1;
		FLOW(1, "work[%d]: ssn<%d> server %d closed\n", 
		     info->index, s->id, fd);
	}

	FLOW(1, "work[%d]: ssn<%d> server %d recv %d bytes\n", 
	     info->index, s->id, fd, n);

	return 0;
}


/**
 *	Parse client data
 *
 *	Return 0 if parse success, -1 on error.
 */
static int 
_work_parse_client(thread_t *info, session_t *s)
{
	packet_t *pkt;
//	int len;
//	void *ptr;

	assert(info);
	assert(info->priv);
	assert(s);

	/* client is closed */
	if (!s->cliq.pkt)
		return 0;

	pkt = s->cliq.pkt;
//	ptr = pkt->data + pkt->recvpos;
//	len = pkt->len - pkt->recvpos;

#if 0
	/* do HTTP parse */
	assert(len >= 0);
	if (len > 0) {
		if (http_parse_request(&s->clihttp, ptr, len, 0)) {
			ERR("work[%d]: http request parse error\n",
			    info->index);
			return -1;
		}
	}
	/* client is closed */
	else {
		if (s->clihttp.state != HTTP_STE_FIN) {
			ERR("work[%d] http request not finish before closed\n",
			    info->index);
			return -1;
		}
	}

	FLOW(1, "work[%d]: ssn<%d> client %d http parse success\n", 
	     info->index, s->id, s->clifd);
	
	/* move packet to output queue */
	if (pkt->len == pkt->capacity ||
	    s->clihttp.state == HTTP_STE_FIN) 
	{
		s->clicache.pkt = NULL;
		pktqueue_in(&s->clicache.output, pkt);

		FLOW(1, "work[%d] ssn<%d> client %d move packet(%p) to output\n", 
		     info->index, s->id, s->clifd, pkt);
	}
#endif 
	s->cliq.pkt = NULL;
	pktq_in(&s->cliq.outq, pkt);

	FLOW(2, "work[%d]: ssn<%d> client %d move packet(%p) to output\n", 
	     info->index, s->id, s->clifd, pkt);

	return 0;
}

/**
 *	Parse server data.
 *
 *	Return 0 if success, -1 on error.
 */
static int
_work_parse_server(thread_t *info, session_t *s)
{
	packet_t *pkt;
//	int len;
//	void *ptr;

	assert(info);
	assert(info->priv);
	assert(s);
	
	/* server is closed */
	if (!s->svrq.pkt)
		return 0;

	pkt = s->svrq.pkt;
//	ptr = pkt->data + pkt->recvpos;
//	len = pkt->len - pkt->recvpos;

#if 0
	if (len > 0) {
		if (http_parse_response(&s->svrhttp, ptr, len, 0)) {
			s->flags |= SESSION_SVRERROR;
			ERR("http response parse error\n");
			return -1;
		}
	}
	/* server is closed */
	else {
		if (s->svrhttp.state != HTTP_STE_FIN) {
			ERR("http response not fin before closed\n");
			return -1;
		}
	}
	
	FLOW(1, "work[%d] ssn<%d> server %d http parse success\n", 
	     info->index, s->id, s->svrfd);
	
	/* move packet to output queue for send */
	if (pkt->len == pkt->capacity ||
	    s->svrhttp.state == HTTP_STE_FIN) 
	{
		s->svrcache.current = NULL;
		pktqueue_in(&s->svrcache.output, pkt);

		FLOW(1, "work[%d] ssn<%d> server %d move packet(%p) to output\n", 
		     info->index, s->id, s->svrfd);	
	}

#endif
	s->svrq.pkt = NULL;
	pktq_in(&s->svrq.outq, pkt);

	FLOW(2, "work[%d]: ssn<%d> server %d move packet(%p) to output\n", 
	     info->index, s->id, s->svrfd, pkt);

	return 0;
}

static int 
_work_sendto_client(thread_t *info, session_t *s)
{
	work_t *wi;
	packet_t *pkt;
	struct epoll_event e;
	void *ptr;
	int *pair;
	int len;
	int n;
	int fd;
	int len1;

	assert(info);
	assert(info->priv);
	assert(s);
	assert(s->clifd > -1);

	wi = info->priv;
	fd = s->clifd;

	if (s->svrq.outq.len < 1 && ! s->svrq.blkpkt)
		return 0;

	/* check block packet first */
	if (s->svrq.blkpkt) {
		pkt = s->svrq.blkpkt;
		s->svrq.blkpkt = NULL;
	}
	else {
		pkt = pktq_out(&s->svrq.outq);
	}

	while (pkt) {
		ptr = pkt->data + pkt->sendpos;
		len = pkt->len - pkt->sendpos;

		len1 = len;
		//if (len > 20) len1 = len - 20;

		n = sk_send(fd, ptr, len1);
		/* send error */
		if (n < 0) {
			s->is_clierror = 1;
			s->nalloced--;
			objpool_put(pkt);
			FLOW(2, "work[%d]: ssn<%d> server %d free a packet %p\n", 
			     info->index, s->id, s->clifd, pkt);
			FLOW(1, "work[%d]: ssn<%d> server %d sendto client %d failed\n", 
			     info->index, s->id, s->svrfd, fd);
			return -1;
		}

		proxy_stat_lock();
		g_proxy.stat.nclisend += n;
		proxy_stat_unlock();

		if (n < len) {
			pkt->sendpos += n;			
			if (s->is_cliblock)
				break;
			
			/* mod it to (write) and set blocked when from nonblocked to blocked */
			e.events = EPOLLOUT;
			pair = (int *)&e.data.u64;
			pair[0] = fd;
			pair[1] = s->id;
			if (epoll_ctl(wi->epfd, EPOLL_CTL_MOD, fd, &e)) {
				objpool_put(pkt);
				s->nalloced--;
				FLOW(2, "work[%d]: ssn<%d> server %d free a packet %p\n", 
				     info->index, s->id, s->svrfd, pkt);
				ERR("epoll mod %d error %s\n", fd, ERRSTR);
				return -1;
			}
			FLOW(3, "work[%d]: ssn<%d> client %d epoll mod read->write\n",
			     info->index, s->id, fd);

			/* remove server fd from epoll */
			if (epoll_ctl(wi->epfd, EPOLL_CTL_DEL, s->svrfd, NULL)) {
				ERR("epoll del %d error %s\n", fd, ERRSTR);
				objpool_put(pkt);
				s->nalloced--;
				FLOW(2, "work[%d]: ssn<%d> server %d free a packet %p\n",
				     info->index, s->id, s->svrfd, pkt);
				ERR("epoll del %d error %s\n", s->svrfd, ERRSTR);
				return -1;
			}
			s->is_svrep = 0;
			FLOW(3, "work[%d]: ssn<%d> server %d epoll del read(blocked)\n",
			     info->index, s->id, s->svrfd);

			s->svrq.blkpkt = pkt;
			s->is_cliblock = 1;
			wi->nblocked++;
			FLOW(1, "work[%d]: ssn<%d> server %d sendto client %d blocked(%d:%d)\n",
			     info->index, s->id, s->svrfd, fd, len, n);
			break;
		}

		if (s->is_cliblock) {
			e.events = EPOLLIN;
			pair = (int *)&e.data.u64;
			pair[0] = fd;
			pair[1] = s->id;
			if (epoll_ctl(wi->epfd, EPOLL_CTL_MOD, fd, &e)) {
				objpool_put(pkt);
				s->nalloced--;
				FLOW(2, "work[%d]: ssn<%d> server %d free a packet %p\n", 
				     info->index, s->id, s->svrfd, pkt);
				ERR("epoll mod %d error %s\n", fd, ERRSTR);
				return -1;
			}
			FLOW(3, "work[%d]: ssn<%d> client %d epoll mod read(not blocked)\n",
			     info->index, s->id, fd);
			s->is_cliblock = 0;
			wi->nblocked--;

			/* add server fd to epoll for read */
			e.events = EPOLLIN;
			pair = (int *)&e.data.u64;
			pair[0] = s->svrfd;
			pair[1] = s->id;
			if (epoll_ctl(wi->epfd, EPOLL_CTL_ADD, s->svrfd, &e)) {
				objpool_put(pkt);
				s->nalloced--;
				FLOW(2, "work[%d]: ssn<%d> server %d free a packet %p\n", 
					info->index, s->id, s->svrfd, pkt);
				ERR("epoll add %d error %s\n", s->svrfd, ERRSTR);
				return -1;
			}
			s->is_svrep = 1;
			FLOW(3, "work[%d]: ssn<%d> server %d epoll add read(not blocked)\n",
			     info->index, s->id, fd);
		}

		FLOW(1, "work[%d]: ssn<%d> server %d sendto client %d %d bytes\n", 
		     info->index, s->id, s->svrfd, s->clifd, n);

		objpool_put(pkt);
		s->nalloced--;

		FLOW(2, "work[%d]: ssn<%d> server %d free packet(%p)\n", 
		     info->index, s->id, s->svrfd, pkt);
		
		pkt = pktq_out(&s->svrq.outq);
	}
	
	return 0;

}

/**
 *	Send client data to server.
 *
 *	Return 0 if success, -1 on error.
 */
static int 
_work_sendto_server(thread_t *info, session_t *s)
{
	work_t *wi;
	packet_t *pkt;
	struct epoll_event event;
	int fd;
	void *ptr;
	int len;
	int n;
	int *pair;
	int len1;

	assert(info);
	assert(info->priv);
	assert(s);

	wi = info->priv;

	if (s->cliq.outq.len < 1 && ! s->cliq.blkpkt)
		return 0;

	if (_work_get_server(info, s)) {
		ERR("work[%d] get server failed\n", info->index);
		return -1;
	}

	if (s->is_svrwait) {
		return 0;
	}

	fd = s->svrfd;

	/* send block packet first */
	if (s->cliq.blkpkt) {
		pkt = s->cliq.blkpkt;
		s->cliq.blkpkt = NULL;
	}
	else {
		pkt = pktq_out(&s->cliq.outq);
	}

	while (pkt) {

		/* send packet to server */
		ptr = pkt->data + pkt->sendpos;
		len = pkt->len - pkt->sendpos;
		len1 = len;
		//if (len > 20) len1 = len - 20;

		n = sk_send(fd, ptr, len1);
		/* send error */
		if (n <= 0) {
			s->is_svrerror = 1;
			objpool_put(pkt);
			s->nalloced--;
			FLOW(2, "work[%d]: ssn<%d> client %d free a packet %p\n", 
			     info->index, s->id, s->clifd, pkt);
			FLOW(1, "work[%d]: ssn<%d> client %d sendto server %d error\n", 
			    info->index, s->id, s->clifd, fd);
			return -1;
		}

		proxy_stat_lock();
		g_proxy.stat.nsvrsend += n;
		proxy_stat_unlock();

		/* send not finished, block client */
		if (n < len) {
			pkt->sendpos = pkt->sendpos + n;
			if (s->is_svrblock)
				break;

			/* mod it to write */
			pair = (int *)&event.data.u64;
			pair[0] = fd;
			pair[1] = s->id;
			event.events = EPOLLOUT;			
			if (epoll_ctl(wi->epfd, EPOLL_CTL_MOD, fd, &event)) {
				s->is_svrerror = 1;
				objpool_put(pkt);
				s->nalloced--;
				FLOW(2, "work[%d]: ssn<%d> client %d free a packet(%p)\n", 
						info->index, s->id, s->clifd, pkt);
				ERR("epoll mod %d error %s\n", fd, ERRSTR);
				return -1;
			}
			FLOW(3, "work[%d]: ssn<%d> server %d epoll mod read->write(blocked)\n",
					info->index, s->id, fd);

			/* delete client fd from epoll */
			if (epoll_ctl(wi->epfd, EPOLL_CTL_DEL, s->clifd, NULL)) {
				ERR("epoll del %d error %s\n", s->clifd, ERRSTR);
				return -1;
			}
			s->is_cliep = 0;
			FLOW(3, "work[%d]: ssn<%d> client %d epoll del read\n",
					info->index, s->id, s->clifd);

			s->cliq.blkpkt = pkt;
			s->is_svrblock = 1;
			wi->nblocked++;
			FLOW(1, "work[%d]: ssn<%d> client %d sendto server %d %d:%d blocked\n", 
			     info->index, s->id, s->clifd, fd, len, n);
			break;
		}

		/* clear block flags because send success */
		if (s->is_svrblock) {
			pair = (int *)&event.data.u64;
			pair[0] = fd;
			pair[1] = s->id;
			event.events = EPOLLIN;			
			if (epoll_ctl(wi->epfd, EPOLL_CTL_MOD, fd, &event)) {
				s->is_svrerror = 1;
				objpool_put(pkt);
				s->nalloced--;
				FLOW(2, "work[%d]: ssn<%d> client %d free a packet(%p)\n", 
						info->index, s->id, s->clifd, pkt);
				ERR("epoll mod %d error %s\n", fd, ERRSTR);
				return -1;
			}
			FLOW(3, "work[%d]: ssn<%d> server %d epoll mod write->read\n",
			     info->index, s->id, fd);
			s->is_svrblock = 0;
			wi->nblocked--;

			/* add client to read epoll */
			pair = (int *)&event.data.u64;
			pair[0] = s->clifd;
			pair[1] = s->id;
			event.events = EPOLLIN;			
			if (epoll_ctl(wi->epfd, EPOLL_CTL_ADD, s->clifd, &event)) {
				s->is_clierror = 1;
				objpool_put(pkt);
				s->nalloced--;
				FLOW(2, "work[%d]: ssn<%d> client %d free a packet(%p)\n", 
						info->index, s->id, s->clifd, pkt);
				ERR("epoll mod %d error %s\n", s->clifd, ERRSTR);
				return -1;
			}
			s->is_cliep = 1;
			FLOW(3, "work[%d]: ssn<%d> client %d epoll add read\n",
			     info->index, s->id, s->clifd);
		}

		FLOW(1, "work[%d]: ssn<%d> client %d sendto server %d %d bytes\n", 
		     info->index, s->id, s->clifd, fd, n);

		s->nalloced--;
		objpool_put(pkt);

		FLOW(2, "work[%d]: ssn<%d> client %d free packet(%p)\n", 
				info->index, s->id, s->svrfd, pkt);
		pkt = pktq_out(&s->cliq.outq);
	}

	return 0;
}

/**
 *	nb_splice client data to server.	
 *
 *	Return nb_splice bytes if success, -1 on error.
 */
static int 
_work_nb_splice_cli2svr(thread_t *info, session_t *s)
{
	int n;

	assert(info);
	assert(s);
	assert(s->clifd > 0);
	assert(s->svrfd > 0);

	if (s->is_svrwait)
		return 0;

	n = nb_splice(g_proxy.nb_splice_fd, s->clifd, s->svrfd, 16384, 0);
	
	FLOW(1, "work[%d] ssn<%d> client %d nb_splice server %d %d bytes\n",
	     info->index, s->id, s->clifd, s->svrfd, n);

	return n;
}

/**
 *	splice(linux) client data to server
 *
 *	Return splice bytes if success, -1 on error.
 */
static int 
_work_splice_cli2svr(thread_t *info, session_t *s)
{
	int n, m;
	
	assert(info);
	assert(info->priv);
	assert(s);
	assert(s->clifd > 0);
	assert(s->svrfd > 0);
	
	if (s->is_svrwait)
		return 0;

	n = splice(s->clifd, NULL, s->write_pipe, NULL, 
		   16384, SPLICE_F_MORE | SPLICE_F_NONBLOCK);
	FLOW(1, "work[%d]: ssn<%d> client %d splice(1) server %d  %d bytes\n", 
	     info->index,s->id, s->clifd, s->svrfd, n);
	if (n <= 0){
		if (n < 0)
			ERR("splice failed: %s\n", strerror(errno));
		return -1;
	}

	m = splice(s->read_pipe, NULL, s->svrfd, NULL, 
		   n, SPLICE_F_MORE | SPLICE_F_NONBLOCK);
	FLOW(1, "work[%d]: ssn<%d> client %d splice(2) server %d %d bytes\n", 
	     info->index, s->id, s->clifd, s->svrfd, m);

	return m;
}

/**
 *	Process client request, include recv client data, parse client data, 
 *	and send client data to server 
 *
 *	Return 0 if success, -1 on error.
 */
static int 
_work_process_client(thread_t *info, session_t *s)
{
	assert(info);
	assert(info->priv);
	assert(s);
	assert(s->clifd >= 0);

	if (g_proxy.use_nb_splice) 
		return _work_nb_splice_cli2svr(info, s);

	if (g_proxy.use_splice) 
		return _work_splice_cli2svr(info, s);

	if (_work_recvfrom_client(info, s)) 
		return -1;
       
	if (_work_parse_client(info, s))
		return -1;
		
	if (_work_sendto_server(info, s)) 
		return -1;

	if (s->is_clierror || s->is_cliclose) 
		return -1;
	
	return 0;
}

/**
 *	nb_splice server data to client.	
 *
 *	Return splice bytes if success, -1 on error.
 */
static int 
_work_nb_splice_svr2cli(thread_t *info, session_t *s)
{
	int n;

	assert(info);
	assert(s);
	assert(s->clifd > 0);
	assert(s->svrfd > 0);

	n = nb_splice(g_proxy.nb_splice_fd, s->svrfd, s->clifd, 16384, 0);

	FLOW(1, "work[%d]: ssn<%d> client %d nb_splice server %d %d bytes\n", 
	     info->index, s->id, s->svrfd, s->clifd, n);

	return n;
}

/**
 *	Splice server data to client.
 *
 *	Return splice bytes is success, -1 on error.
 */
static int 
_work_splice_svr2cli(thread_t *info, session_t *s)
{
	int n, m;
	
	assert(info);
	assert(s);
	assert(s->clifd > 0);
	assert(s->svrfd > 0);
	
	n = splice(s->svrfd, NULL, s->write_pipe, NULL, 16384, SPLICE_F_NONBLOCK);
	FLOW(1, "work[%d]: ssn<%d> server %d splice(1) client %d %d bytes\n", 
	     info->index, s->id, s->svrfd, s->clifd, n);
	if (n <= 0) {
		if (n < 0)
			ERR("splice failed %s\n", strerror(errno));
		return -1;
	}

	m = splice(s->read_pipe, NULL, s->clifd, NULL, n, SPLICE_F_NONBLOCK);
	FLOW(1, "work[%d]: ssn<%d> server %d splice(2) client %d %d bytes\n", 
	     info->index, s->id, s->svrfd, s->clifd, m);

	return m;
}

/**
 *	Process server data, include recv data data, parse server data, and
 *	send server data to client.
 *
 *	Return 0 if success, -1 on error.
 */
static int
_work_process_server(thread_t *info, session_t *s)
{
	assert(info);
	assert(info->priv);
	assert(s);
	assert(s->svrfd > -1);

	if (g_proxy.use_nb_splice) 
		return _work_nb_splice_svr2cli(info, s);

	if (g_proxy.use_splice) 
		return _work_splice_svr2cli(info, s);

	if (_work_recvfrom_server(info, s)) 
		return -1;

	if (_work_parse_server(info, s)) 
		return -1;

	if (_work_sendto_client(info, s)) 
		return -1;

	if (s->is_svrerror || s->is_svrclose) 
		return -1;

	return 0;
}

/**
 *	Handle epoll event.
 *
 * 	Return 0 if success, -1 on error.
 */
static int 
_work_epoll(thread_t *info)
{
	work_t *wi;
	session_t *s;
	struct epoll_event *e;
	int *pair;
	int fd;
	int id;
	int nfds;
	int i;

	assert(info);
	assert(info->priv);

	wi = info->priv;

	nfds = epoll_wait(wi->epfd, wi->events, 
			  wi->nevent, WORK_TIMEOUT);
	if (nfds < 0) {
		if (errno == EINTR)
			return 0;
		ERR("epoll_wait error %s", ERRSTR);
		return -1;
	}

	if (nfds == 0)
		return -1;

	for (i = 0; i < nfds; i++) {
		
		e = &wi->events[i];

		pair = (int *)&e->data.u64;
		fd = pair[0];
		id = pair[1];

		if (fd == wi->httpfd) {
			if (e->events & EPOLLERR) {
				ERR("work[%d] listen fd %d epoll error: %s\n", 
				    info->index, fd, ERRSTR);
				continue;
			}			
			_work_accept_client(info, fd);
			continue;
		}

		s = session_table_find(wi->sesstbl, id);
		if (!s) {
			continue;
		}

		/* check it's listen fd have new client */
		if (e->events & EPOLLERR) {
			FLOW(1, "work[%d]: ssn<%d> fd %d epoll error\n", 
			     info->index, id, fd);
			_work_free_session(info, s);
			continue;
		}

		if (s->clifd == fd) {
			if (e->events & EPOLLIN) {
				FLOW(3, "work[%d]: ssn<%d> client %d have read event\n", 
				     info->index, s->id, fd);
				if (_work_process_client(info, s)) {
					_work_free_session(info, s);
				}
			}
			else if (e->events & EPOLLOUT) {
				FLOW(3, "work[%d]: ssn<%d> client %d have write event\n", 
				     info->index, s->id, fd);
				if (_work_sendto_client(info, s)) {
					_work_free_session(info, s);
				}
			}
			else {
				ERR("work[%d]: ssn<%d> client %d unknow event %d\n", 
				    info->index, id, fd, e->events);
			}
		}
		else if (s->svrfd == fd) {
			if (e->events & EPOLLIN) {
				FLOW(3, "work[%d]: ssn<%d> server %d have read event\n", 
				     info->index, s->id, fd);
				if (_work_process_server(info, s)) {
					_work_free_session(info, s);
				}
			}
			else if (e->events & EPOLLOUT) {
				FLOW(3, "work[%d]: ssn<%d> server %d have write event\n", 
				     info->index, s->id, fd);
				if (s->is_svrwait && _work_check_server(info, s)) {
					_work_free_session(info, s);
					continue;
				}

				if (_work_sendto_server(info, s)) {
					_work_free_session(info, s);
				}
			}
			else {
				ERR("work[%d] ssn<%d> server %d unknow event %d\n", 
				    info->index, id, fd, e->events);
			}
		}
		else {
			ERR("the wrong fd %d is in session %d\n", fd, id);
		}
	}

	return 0;
}

/**
 *	work thread main loop
 *	
 *	Always return 0.
 */
static int 
_work_loop(thread_t *info)
{
	while (!g_proxy.stop) {
		_work_epoll(info);
	}

	return 0;
}

/**
 *	The main function of work thread. 
 *
 *	Return NULL always.
 */
void *
work_run(void *arg)
{
	thread_t *info;
	int cpu;

	info = arg;
	assert(info);

	DBG("work[%d] started\n", info->index);

	/* bind to cpu */
	if (g_proxy.bind_cpu) {
		cpu = thread_bind_cpu(pthread_self(), info->index, 
				      g_proxy.bind_cpu_algo, 
				      g_proxy.bind_cpu_ht);
		if (cpu < 0) {
			ERR("work[%d] bind CPU failed", info->index);
			pthread_exit(0);
		}

		printf("work[%d] bind to cpu %d\n", info->index, cpu);
	}

	if (_work_init(info)) {
		if (g_proxy.main_tid > 0) {
			pthread_kill(g_proxy.main_tid, SIGINT);
		}
		pthread_exit(0);
	}

	_work_loop(info);

	_work_free(info);

	DBG("work[%d] stoped\n", info->index);

	pthread_exit(0);
}


